package com.youlin.untils;

import com.youlin.aop.DelayMessagePostProcessor;
import org.redisson.api.RBloomFilter;
import org.redisson.api.RLock;
import org.redisson.api.RedissonClient;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.data.redis.core.RedisTemplate;
import org.springframework.jms.core.JmsTemplate;
import org.springframework.stereotype.Component;
import javax.jms.Topic;
import java.lang.ref.SoftReference;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.TimeUnit;
import java.util.function.Supplier;

@Component
public class MyCacheUtil {

    @Autowired
    private RedissonClient redissonClient;
    @Autowired
    private JmsTemplate jmsTemplate;
    @Autowired
    private RedisTemplate<String, Object> redisTemplate;
    @Autowired
    private Topic dmLocalCacheTopic;
    @Autowired
    private Topic amStatusLocalCacheTopic;
    @Autowired
    private Topic inlLocalCacheTopic;

    private static <T> T getFromCache(ConcurrentHashMap<String, SoftReference<T>> concurrentHashMap, String key) {
        SoftReference<T> tSoftReference = concurrentHashMap.get(key);
        if (tSoftReference == null) return null;
        else return tSoftReference.get();
    }

    private <T> T putIntoCache(ConcurrentHashMap<String, SoftReference<T>> concurrentHashMap, String finalCacheKey, T cache) {
        concurrentHashMap.put(finalCacheKey, new SoftReference<>(cache));
        String[] keys = finalCacheKey.split(":");
        Topic destination;
        switch (keys[0]) {
            case "amUuId_dm":
                destination = dmLocalCacheTopic;
                break;
            case "amUuId_status":
                destination = amStatusLocalCacheTopic;
                break;
            case "amUuId_isNL":
                destination = inlLocalCacheTopic;
                break;
            default:
                destination = null;
        }
        if (destination != null) {
            Boolean setIfAbsent = redisTemplate.opsForValue().setIfAbsent(
                    "cleanLocalCache:" + finalCacheKey, 123, 1000*60*5L, TimeUnit.MILLISECONDS);
            if (setIfAbsent) jmsTemplate.convertAndSend(destination, finalCacheKey, new DelayMessagePostProcessor(1000*60*5L));
        }
        return cache;
    }

    public static <T> void delInCache(ConcurrentHashMap<String, SoftReference<T>> concurrentHashMap, String key) {
        concurrentHashMap.remove(key);
    }

    public <T> T getFromCache(ConcurrentHashMap<String, SoftReference<T>> cacheMap, MyEnum cacheKeyPrefix, String cacheKey,
            MyEnum bloomKey, Class<T> cacheClass, MyEnum lockKey, Supplier<T> cacheSupplier, Long EXPIRELONG, TimeUnit TIMEUNIT) {
        String finalCacheKey = cacheKeyPrefix.getMessage() + cacheKey;
//        System.out.printf("\033[34m ===>> 首先从本地缓存中获取 <<=== \033[0m\n");
        T t = MyCacheUtil.getFromCache(cacheMap, finalCacheKey);
        if (t == null) {
            // 布隆过滤器：判断一个key不存在无误差，判断一个key存在则有一定误差(误差率，可配置，不可消除)
            if (bloomKey != null) {
                RBloomFilter<Object> bloomFilter = redissonClient.getBloomFilter(bloomKey.getMessage());
                if (! bloomFilter.contains(cacheKey)) {
//                System.out.printf("\033[34m ===>> 分布式布隆过滤器防止缓存穿透 <<=== \033[0m\n");
                    throw new ParamErrorExpectation();
                }
            }
//            System.out.printf("\033[34m ===>> 再次从 Redis 缓存中获取 <<=== \033[0m\n");
            t = RedisUtil.getObject(redisTemplate, finalCacheKey, cacheClass);
            if (t == null) {
                // 获取 Redis 分布式锁
                RLock rLock = redissonClient.getLock(lockKey.getMessage());
                try {
                    rLock.lock();
//                    System.out.printf("\033[34m ===>> 双端检索 从本地缓存中获取 <<=== \033[0m\n");
                    t = MyCacheUtil.getFromCache(cacheMap, finalCacheKey);
                    if (t == null) {
//                        System.out.printf("\033[34m ===>> 双端检索 从 Redis 缓存中获取 <<=== \033[0m\n");
                        t = RedisUtil.getObject(redisTemplate, finalCacheKey, cacheClass);
                        if (t == null) {
//                            System.out.printf("\033[34m ===>> 最后从数据库中获取 <<=== \033[0m\n");
                            t = cacheSupplier.get();
                            if (t != null) {
                                if (EXPIRELONG != null && TIMEUNIT != null)
                                    RedisUtil.setObjectExpire(redisTemplate, finalCacheKey, t, EXPIRELONG, TIMEUNIT);
                                else RedisUtil.setObjectExpire(redisTemplate, finalCacheKey, t);
                                return putIntoCache(cacheMap, finalCacheKey, t);
                            } else return t;
                        } else return putIntoCache(cacheMap, finalCacheKey, t);
                    } else return t;
                } finally {
                    rLock.unlock();
                }
            } else return putIntoCache(cacheMap, finalCacheKey, t);
        } else return t;
    }

    public <T> T getFromCache(ConcurrentHashMap<String, SoftReference<T>> cacheMap, MyEnum cacheKeyPrefix, String cacheKey,
                                            MyEnum bloomKey, Class<T> cacheClass, MyEnum lockKey, Supplier<T> cacheSupplier) {
        return getFromCache(cacheMap, cacheKeyPrefix, cacheKey, bloomKey, cacheClass,
                lockKey, cacheSupplier, 5L, TimeUnit.MINUTES);
    }

}
